package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TradeProvinceOrderBean;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.function.AsyncDimFunctionHBase;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/5/5 14:31
 */
public class Dws_10_DwsTradeProvinceOrderWindow_Doris extends BaseApp {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow_Doris().init(
            40009,
            2,
            "Dws_10_DwsTradeProvinceOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TradeProvinceOrderBean> beanStream = stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getString("order_id")) // 分组: 用来统计订单数
            .process(new KeyedProcessFunction<String, JSONObject, TradeProvinceOrderBean>() {
                
                private ValueState<Boolean> isFirstState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<Boolean> stateDesc = new ValueStateDescriptor<>("isFirst", Boolean.class);
                    StateTtlConfig config = StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(10))
                        .updateTtlOnReadAndWrite()
                        .useProcessingTime()
                        //.returnExpiredIfNotCleanedUp()
                        .build();
                    stateDesc.enableTimeToLive(config);
                    isFirstState = getRuntimeContext().getState(stateDesc);
                }
                
                @Override
                public void processElement(JSONObject obj,
                                           Context ctx,
                                           Collector<TradeProvinceOrderBean> out) throws Exception {
                    Long orderCount = 0L;
                    Boolean isFirst = isFirstState.value();
                    if (isFirst == null) {
                        orderCount = 1L;
                        isFirstState.update(false);
                    }
                    
                    out.collect(TradeProvinceOrderBean.builder()
                                    .orderDetailId(obj.getString("id"))
                                    .provinceId(obj.getString("province_id"))
                                    .orderCount(orderCount)
                                    .orderAmount(obj.getBigDecimal("split_total_amount"))
                                    .ts(obj.getLong("ts") * 1000)
                                    .build());
                    
                }
            })
            .keyBy(TradeProvinceOrderBean::getOrderDetailId)
            .process(new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                
                private ValueState<Boolean> isFirstState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<Boolean> stateDesc = new ValueStateDescriptor<>("isFirst", Boolean.class);
                    StateTtlConfig config = StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(10))
                        .updateTtlOnReadAndWrite()
                        .useProcessingTime()
                        //.returnExpiredIfNotCleanedUp()
                        .build();
                    stateDesc.enableTimeToLive(config);
                    isFirstState = getRuntimeContext().getState(stateDesc);
                }
                
                @Override
                public void processElement(TradeProvinceOrderBean value,
                                           Context ctx,
                                           Collector<TradeProvinceOrderBean> out) throws Exception {
                    // 只需要第一个详情: 因为所有需要的数据都在左表
                    if (isFirstState.value() == null) {
                        isFirstState.update(false);
                        out.collect(value);
                    }
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeProvinceOrderBean::getProvinceId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1,
                                                         TradeProvinceOrderBean value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String provinceId,
                                        Context context,
                                        Iterable<TradeProvinceOrderBean> elements,
                                        Collector<TradeProvinceOrderBean> out) throws Exception {
                        TradeProvinceOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.toPartitionDate(context.window().getStart()));
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                        
                    }
                }
            );
        
AsyncDataStream
    .unorderedWait(
        beanStream,
        new AsyncDimFunctionHBase<TradeProvinceOrderBean>() {
            @Override
            public String getTable() {
                return "dim_base_province";
            }
            
            @Override
            public String getId(TradeProvinceOrderBean bean) {
                return bean.getProvinceId();
            }
            
            @Override
            public void addDim(TradeProvinceOrderBean bean, JSONObject dim) {
                bean.setProvinceName(dim.getString("name"));
            }
        },
        60,
        TimeUnit.SECONDS
    )
    .map(bean -> {
        SerializeConfig config = new SerializeConfig();
        config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
        return JSON.toJSONString(bean, config);
    })
    .sinkTo(DorisUtil.getDorisSink("gmall.dws_trade_province_order_window"));
        
        
    }
}
/*
dwd 层下单事务事实表

order_id  order_detail_id   订单数  订单金额
1				1			 1       100
1				2            0       200
1				3			 0       300


1. 读取下单事务事实表

2. 解析成 pojo 类型
	按照 order_id 分组,  把这个订单的第一个详情中的订单数置为 1,其他为 9

3. 按照详情 id 去重
	由于下单金额只在左表, 只需要取这个详情的第一条数据就行了

4. 开窗聚合

5. 补充维度
	省份的名字

6. 写出到 clickhouse 中
 */